package com.derbysoft.nuke.kafka.manager.web.controller;

import com.derbysoft.nuke.kafka.manager.KafkaManagerService;
import com.derbysoft.nuke.kafka.manager.domain.model.*;
import com.derbysoft.nuke.kafka.manager.domain.service.ClusterService;
import com.derbysoft.nuke.kafka.manager.dto.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.util.List;
import java.util.Map;

/**
 * Created by Albert Fu on 16/05/2018.
 */
@RestController
public class ClusterController {

    @Autowired
    private ClusterService clusterService;

    @Autowired
    private KafkaManagerService kafkaManagerService;

    @PostMapping(value = "/cluster/save")
    public void saveCluster(@RequestBody Cluster cluster) {
        if (cluster.getCode().equals("error")) {
            throw new IllegalArgumentException("XXX");
        }
        clusterService.saveCluster(cluster);
    }

    @GetMapping(value = "/cluster/{clusterCode}/delete")
    public void deleteCluster(@PathVariable("clusterCode") String clusterCode) {
        clusterService.deleteCluster(clusterCode);
    }

    @PostMapping(value = "/cluster/batch/delete")
    public void deleteClusters(@RequestParam(value = "codes") String[] codes) {
        clusterService.deleteClusters(codes);
    }

    @GetMapping(value = "/clusters")
    public List<Cluster> getClusters() {
        return clusterService.getClusters();
    }

    @GetMapping(value = "/cluster/{clusterCode}")
    public Cluster getCluster(@PathVariable("clusterCode") String clusterCode) {
        return clusterService.getClusterByCode(clusterCode);
    }

    @PostMapping(value = "/cluster/{clusterCode}/topics")
    public Topic createTopic(
            @PathVariable("clusterCode") String clusterCode,
            @RequestBody Map<String, Object> topic) {
        throw new UnsupportedOperationException("not yet implement");
    }

    @GetMapping(value = "/cluster/{clusterCode}/topics")
    public List<TopicSummary> getClusterTopics(@PathVariable("clusterCode") String clusterCode) {
        Cluster cluster = clusterService.getClusterByCode(clusterCode);
        return kafkaManagerService.listTopicSummaries(cluster);
    }

    @GetMapping(value = "/cluster/{clusterCode}/topic/{topicName}")
    public Topic getClusterTopic(@PathVariable("clusterCode") String clusterCode, @PathVariable("topicName") String topicName) {
        Cluster cluster = clusterService.getClusterByCode(clusterCode);
        return kafkaManagerService.getTopic(cluster, topicName);
    }

    @GetMapping(value = "/cluster/{clusterCode}/topic/{topicName}/partition/summaries")
    public List<TopicPartitionSummary> getTopicPartitionSummaries(@PathVariable("clusterCode") String clusterCode, @PathVariable("topicName") String topicName) {
        Cluster cluster = clusterService.getClusterByCode(clusterCode);
        return kafkaManagerService.getTopicPartitionSummaries(cluster, topicName);
    }

    @PostMapping(value = "/cluster/{clusterCode}/topic/{topicName}/message/seek")
    public TopicMessageSeekResponse seekTopicMessages(
            @PathVariable("clusterCode") String clusterCode,
            @PathVariable("topicName") String topicName,
            @RequestBody TopicMessageSeekRequest topicMessageSeekRequest
    ) {
        Cluster cluster = clusterService.getClusterByCode(clusterCode);
        return kafkaManagerService.seekMessages(cluster, topicName, topicMessageSeekRequest);
    }

    @GetMapping(value = "/cluster/{clusterCode}/brokers")
    public List<Broker> getClusterBrokers(@PathVariable("clusterCode") String clusterCode) {
        Cluster cluster = clusterService.getClusterByCode(clusterCode);
        return kafkaManagerService.getBrokers(cluster);
    }

    @GetMapping(value = "/cluster/{clusterCode}/brokers/metrics")
    public List<BrokerMetrics> getClusterBrokersMetrics(@PathVariable("clusterCode") String clusterCode) {
        Cluster cluster = clusterService.getClusterByCode(clusterCode);
        return kafkaManagerService.getBrokersMetrics(cluster);
    }

}
